/*
 * Copyright 2009 BioTeam Inc
 * 
 * Licensed under License GNU LESSER GENERAL PUBLIC LICENSE version 3.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 * 
 * http://www.gnu.org/copyleft/lesser.html
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package net.bioteam.appweb.submit;

import java.text.SimpleDateFormat;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.TimerTask;

import javax.mail.MessagingException;
import javax.mail.internet.AddressException;
import javax.persistence.EntityManager;
import javax.persistence.EntityTransaction;
import javax.persistence.PersistenceException;
import javax.persistence.Query;

import net.bioteam.appweb.Config;
import net.bioteam.cloud.cluster.ActiveCluster;
import net.bioteam.cloud.cluster.ClusterStatus;
import net.bioteam.common.utils.MailSender;
import net.bioteam.common.utils.MyEntityManager;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * Update job status submit pending job.
 * 
 * @author jason
 * 
 */
public class JobCollector extends TimerTask
{
	public static Log logger = LogFactory.getLog("net.bioteam.appweb.utils.StatusChecker");
	public static RunStatusChecker checker;

	public static String URLPREFIX = "";

	// delay in millisecond
	public static int interval = 60000;

	public void run()
	{

		EntityManager em = null;
		MailSender sender = null;

		try
		{
			em = MyEntityManager.getEntityManager();
		} catch (Exception e)
		{
			logger.warn(e.getMessage());
			return;
		}
		// load object out of the transaction so that all of these objects are
		// not managed
		// this is not true for hibernate
		try
		{
			Query query = em.createNamedQuery("allunfinishedjobs");
			List<UnfinishedJob> jobs = (List<UnfinishedJob>) query.getResultList();

			if (jobs.isEmpty())
			{
				return;
			}

			if (MailSender.session != null)
			{
				sender = new MailSender();
				try
				{
					sender.openTransport();
				} catch (MessagingException e)
				{
					logger.warn("error when open mail session transport " + e.getMessage(), e);

					sender = null;
				}
			}
			Map<Long, List<UnfinishedJob>> cjobs = clusterAndJob(jobs);
			
			
			Map<Long, Boolean> touchedACs=new HashMap<Long, Boolean>();
			Query query1 = em.createNamedQuery("getallactive");
			List<ActiveCluster> acs =(List<ActiveCluster>) query1.getResultList();
			

			for (UnfinishedJob job : jobs)
			{
				try
				{
					if (job.isRemoteJob())
					{
						handleRemoteJob(job, sender, em, cjobs.get(job.getClusterid()));
						touchedACs.put(job.getClusterid(), true);
					} else
					{
						handleLocalJob(job, sender, em);
					}
				} catch (Exception e)
				{
					EntityTransaction tx = em.getTransaction();
					try
					{
						if (tx.isActive())
							tx.rollback();
					} catch (Exception e1)
					{

					}
					logger.warn(e.getMessage());
					continue;
				}
			}
			
			EntityTransaction tx=em.getTransaction();
			tx.begin();
			for (ActiveCluster ac: acs)
			{
				if (touchedACs.containsKey(ac.getId()))
					continue;
				else
				{
					//no job is associated with this cluster anymore.
					ac.setStatus(ClusterStatus.IDLE);
					ac.setNote("no job is associated with this cluster");
				}
			}
			tx.commit();
			

		} catch (Exception e2)
		{
			logger.warn(e2.getMessage());
		} finally
		{
			try
			{

				MyEntityManager.close();
				if (sender != null)
					sender.closeTransport();
			} catch (Exception e1)
			{
				// ignore
			}

		}
	}

	protected void handleLocalJob(UnfinishedJob job, MailSender sender, EntityManager em)
	{
		EntityTransaction tx = em.getTransaction();
		switch (job.getCode())
		{
		case SUBMITTED:
		case RUNNING:
			try
			{
				checker.checkStatus(job);
				checker.postJobProcess(job);
			} catch (Exception e)
			{

				logger.warn("checkstatus exception", e);
				return;
			} finally
			{
				// have status committed to database.
				tx.begin();
				em.flush();
				tx.commit();
			}
			if (!job.isRunning())
			{
				// ---send email
				if (sender != null)
				{
					try
					{
						sendJobMessage(job, sender);
					} catch (Exception e)
					{
						logger.warn("error when sending job finish email to " + job.getEmail() + " " + e.getMessage(), e);
						try
						{
							sender.closeTransport();
						} catch (MessagingException e1)
						{
							// ignore
						}
						sender = null;
					}
				}
			}
			break;
		default:
			tx.begin();
			em.remove(job);
			tx.commit();
			break;
		}
	}

	protected void handleRemoteJob(UnfinishedJob job, MailSender sender, EntityManager em, List<UnfinishedJob> jobs)
	{
		EntityTransaction tx = em.getTransaction();
		switch (job.getCode())
		{
		case PRESUBMIT:
		case FAILEDSUBMISSION:
			if (job.getTrials() < Config.JOB_TRIAL_NUM)
			{
				try
				{
					// once submitted, the cluster status is set to running.
					SubmitterUtil.submitJob(job);
				} catch (PersistenceException e)
				{
					throw e;
				} catch (Exception e)
				{
					logger.warn("error when sending job finish email to " + job.getEmail() + " " + e.getMessage(), e);
					return;
				}
			} else
			{
				if (job.getCode() != RunStatusCode.FAILEDSUBMISSION)
				{
					job.setCode(RunStatusCode.FAILEDSUBMISSION);
					tx.begin();
					em.flush();
					tx.commit();
				}
				finishRemoteJob(job, jobs);
			}
			break;
		case SUBMITTED:
		case RUNNING:
			try
			{
				SoapStatusChecker tempChecker = new SoapStatusChecker();
				// cluster status is set to retrieving.
				tempChecker.checkStatus(job);
				// cluster status is not set by this. final status is set by
				// finishRemoteJob
				tempChecker.postJobProcess(job);
			} catch (Exception e)
			{
				logger.warn("checkstatus exception", e);
				return;
			} finally
			{
				tx.begin();
				em.flush();
				tx.commit();
			}
			break;
		case RETRIEVING:
			try
			{
				SoapStatusChecker tempChecker = new SoapStatusChecker();
				tempChecker.postJobProcess(job);
			} catch (Exception e)
			{
				logger.warn("checkstatus exception", e);
				return;
			}
			break;
		case FINISHED:
		case ABORTED:
		case UNKNOWN:

			// ---send email
			if (sender != null)
			{
				try
				{
					sendJobMessage(job, sender);
				} catch (Exception e)
				{
					logger.warn("error when sending job finish email " + e.getMessage(), e);
					try
					{
						sender.closeTransport();
					} catch (MessagingException e1)
					{
						// ignore
					}
					sender = null;
				}
			}

			finishRemoteJob(job, jobs);
			break;
		}
	}

	protected void finishRemoteJob(UnfinishedJob job, List<UnfinishedJob> jobs)
	{
		EntityManager em = MyEntityManager.getEntityManager();
		EntityTransaction tx = em.getTransaction();
		tx.begin();
		ActiveCluster ac = em.find(ActiveCluster.class, job.getClusterid());
		if (ac != null)
		{
			ClusterStatus s = decideClusterStatus(jobs);
			if (s != ac.getStatus())
				ac.setStatus(s);
		}
		em.remove(job);
		tx.commit();
	}

	protected Map<Long, List<UnfinishedJob>> clusterAndJob(List<UnfinishedJob> jobs)
	{
		Map<Long, List<UnfinishedJob>> ret = new LinkedHashMap<Long, List<UnfinishedJob>>();
		for (UnfinishedJob job : jobs)
		{
			if (job.isRemoteJob())
			{
				if (ret.containsKey(job.getClusterid()))
				{
					ret.get(job.getClusterid()).add(job);
				} else
				{
					List<UnfinishedJob> list = new LinkedList<UnfinishedJob>();
					list.add(job);
					ret.put(job.getClusterid(), list);
				}
			}
		}
		return ret;
	}

	/**
	 * cluster is perform or is ready to perform some task
	 * 
	 * @param jobs
	 * @return
	 */
	protected ClusterStatus decideClusterStatus(List<UnfinishedJob> jobs)
	{
		boolean runningJob = false;
		boolean waitingForSubmit = false;
		boolean retrieving = false;
		for (UnfinishedJob job : jobs)
		{
			switch (job.getCode())
			{
			case PRESUBMIT:
			case SUBMITTED:
				waitingForSubmit = true;
				break;
			case RUNNING:
				runningJob = true;
				break;
			case RETRIEVING:
				retrieving = true;
				break;
			case FAILEDSUBMISSION:
				if (job.getTrials() < Config.JOB_TRIAL_NUM)
					waitingForSubmit = true;
				break;

			}
		}
		if (retrieving)
			return ClusterStatus.RETRIEVING;
		if (runningJob)
			return ClusterStatus.RUNNING;
		if (waitingForSubmit)
			return ClusterStatus.STARTED;
		return ClusterStatus.IDLE;
	}
	
	
	private void sendJobMessage(UnfinishedJob job, MailSender sender) throws AddressException, MessagingException
	{
		StringBuffer sb=new StringBuffer();
		sb.append("URL: "+URLPREFIX+job.getId()+"\n");
		sb.append("Job ID: "+job.getId()+"\n");
		sb.append("Application Name: "+job.getProgramName()+"\n");
		sb.append("Cluster ID: "+(job.getClusterid()== RunStatus.LOCAL_CLUSTER?"Local":job.getClusterid())+"\n");
		sb.append("Command Line: "+ job.getCmdline()+"\n");
		SimpleDateFormat fmt = new SimpleDateFormat("yyyy MM dd HH:mm:ss");
		sb.append("Submit Time: "+(job.getSubmittedTime() == null ? "" : fmt.format(job.getSubmittedTime()))+"\n");
		sb.append("Finish Time: "+(job.getFinishedTime() == null ? "" : fmt.format(job.getFinishedTime()))+"\n");
		sb.append("Note: "+(job.getNote()==null?"":job.getNote())+"\n");
		sb.append("workdir: "+job.getWorkdirName()+"\n");
		sender.sendMessag(Config.ADMIN_EMAIL, job.getEmail(), sb.toString(), "Your job " + job.getProgramName() + " completed");	
	}
}
